#pragma once
#include "../common/net.hpp"
#include "../common/message.hpp"
#include <unordered_set>

namespace myrpc {
    namespace server {
        // 服务端的主题管理
        class TopicManager {
        public:
            using ptr = std::shared_ptr<TopicManager>;
            TopicManager(){}
        private:
            // 订阅者类
            struct Subscriber { 
                using ptr = std::shared_ptr<Subscriber>;
                Subscriber(const BaseConnection::ptr &conn) :_conn(conn) {}
                // 订阅者增加新的订阅主题
                void addTopic(const std::string &topic_name){
                    std::unique_lock<std::mutex> lock(_mutex);
                    _topic_names.insert(topic_name);
                }
                // 订阅者删除订阅主题
                void removeTopic(const std::string &topic_name){
                    std::unique_lock<std::mutex> lock(_mutex);
                    _topic_names.erase(topic_name);
                }

                std::mutex _mutex;
                BaseConnection::ptr _conn;                                   // 订阅者对应的连接
                std::unordered_set<std::string> _topic_names;                // 订阅的所有主题名称
            };
            // 主题类
            struct Topic {
                using ptr = std::shared_ptr<Topic>;
                Topic(const std::string &name):_topic_name(name) {}
                // 主题增加新的订阅者
                void addSubscriber(const Subscriber::ptr &subscriber){
                    std::unique_lock<std::mutex> lock(_mutex);
                    _subscriber_obj.insert(subscriber);
                }
                // 主题删除订阅者
                void removeSubscriber(const Subscriber::ptr &subscriber){
                    std::unique_lock<std::mutex> lock(_mutex);
                    _subscriber_obj.erase(subscriber);
                }
                // 主题发布订阅消息
                void pushMessage(const BaseMessage::ptr &msg){
                    std::unique_lock<std::mutex> lock(_mutex);
                    for(auto &subscriber_conn : _subscriber_obj){
                        subscriber_conn->_conn->send(msg);
                    }
                }

                std::mutex _mutex;
                std::string _topic_name;                                      // 主题名称
                std::unordered_set<Subscriber::ptr> _subscriber_obj;        // 订阅该主题的所有订阅者对象
            };
 
        public:
            // 主题请求的回调处理——注册到Dispatcher消息分发器中 
            void onTopicRequest(const BaseConnection::ptr &conn, TopicRequest::ptr &top_req){
                TopicOpType optype = top_req->getTopicOpType();
                std::string topic_name = top_req->getTopicKey();
                server_logger->INFO("收到主题操作请求: %d, 主题: '%s'", static_cast<int>(optype), topic_name.c_str());
                bool ret = true;
                switch (optype){
                    case TopicOpType::TOPIC_CREATE:
                        topicCreate(conn, top_req);
                        break;
                    case TopicOpType::TOPIC_REMOVE:
                        topicRemove(conn, top_req); 
                        break;
                    case TopicOpType::TOPIC_SUBSCRIBE:
                        ret = topicSubscrible(conn, top_req);
                        break;
                    case TopicOpType::TOPIC_CANCEL:
                        topicCancel(conn, top_req);
                        break;
                    case TopicOpType::TOPIC_PUBLISH:
                        ret = topicPublish(conn, top_req);
                        break;
                    default:
                        // 主题操作不存在
                        server_logger->ERROR("收到无效的主题操作类型: %d", static_cast<int>(optype));
                        return errorResponse(conn, top_req, RCode::RCODE_INVALID_OPTYPE);
                }
                // 结果出错 只有主题不存在一种错误
                if (ret == false){
                    return errorResponse(conn, top_req, RCode::RCODE_NOT_FOUND_TOPIC);
                }
                return topicResponse(conn, top_req, RCode::RCODE_OK);
            }
            // 连接关闭的回调处理——注册到Dispatcher消息分发器中 
            void onConnshutdown(const BaseConnection::ptr &conn){
                std::vector<Topic::ptr> topics;
                Subscriber::ptr subscribe;
                // 1.根据连接查找是否是订阅者关联的连接关闭
                //   不是订阅者关联的连接关闭直接返回 是先获取订阅者对象 
                //   是先获取订阅者对象 然后依次获取订阅的所有主题对象并删除连接和订阅者的映射关系
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto sub_it = _subscribers.find(conn);
                    if (sub_it == _subscribers.end()){
                        // 不是订阅者关联的连接关闭 直接返回
                        server_logger->DEBUG("一个非订阅者连接关闭");
                        return;
                    }
                    subscribe = sub_it->second;
                    server_logger->INFO("订阅者连接关闭, 涉及 %zu 个主题", subscribe->_topic_names.size());
                    for(auto &topic_name : subscribe->_topic_names){
                        auto topic_it = _topics.find(topic_name);
                        if (topic_it == _topics.end()){
                            continue;
                        }
                        topics.push_back(topic_it->second);
                    }
                    _subscribers.erase(sub_it);
                }
                // 2.依次遍历所有的主题对象 删除连接关闭的订阅者对象
                for(auto &topic : topics){
                    topic->removeSubscriber(subscribe);
                }
            }

        private:
            // 主题创建
            void topicCreate(const BaseConnection::ptr &conn, const TopicRequest::ptr &top_req){
                std::unique_lock<std::mutex> lock(_mutex);
                std::string topic_name = top_req->getTopicKey();
                server_logger->DEBUG("创建主题: '%s'", topic_name.c_str());
                auto topic = std::make_shared<Topic>(topic_name);
                _topics.insert(std::make_pair(topic_name, topic));
            }
            // 主题删除
            void topicRemove(const BaseConnection::ptr &conn, const TopicRequest::ptr &top_req){
                // 1.根据主题先找到主题对象 然后获取该主题所有关联的订阅者连接
                //   根据连接找到所有的订阅者对象
                std::string topic_name=top_req->getTopicKey();
                Topic::ptr topic;
                std::unordered_set<Subscriber::ptr> subscribers;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it_top = _topics.find(topic_name);
                    if (it_top == _topics.end()){
                        server_logger->ERROR("主题删除失败: 主题 '%s' 不存在", topic_name.c_str());
                        return;
                    }
                    topic = it_top->second;
                    subscribers = topic->_subscriber_obj;
                    _topics.erase(it_top);
                }
                // 2.依次遍历获取到的订阅者对象 删除主题
                for (auto &subscriber : subscribers){
                    subscriber->removeTopic(topic_name);
                }
            }
            // 主题订阅
            bool topicSubscrible(const BaseConnection::ptr &conn, const TopicRequest::ptr &top_req){
                // 1.根据订阅的主题先找到主题对象 以及订阅者对象
                //   如果没有找到主题对象——报错 如果没有找到订阅者对象——创建一个新的订阅者
                std::string topic_name=top_req->getTopicKey();
                Topic::ptr topic;
                Subscriber::ptr subscriber;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topics.find(topic_name);
                    if (topic_it == _topics.end()){
                        server_logger->ERROR("订阅失败: 主题 '%s' 不存在", topic_name.c_str());
                        return false;
                    }
                    topic = topic_it->second;
                    auto sub_it = _subscribers.find(conn);
                    if (sub_it == _subscribers.end()){
                        subscriber = std::make_shared<Subscriber>(conn);
                        _subscribers.insert(std::make_pair(conn, subscriber));
                    }
                    else{
                        subscriber = sub_it->second;
                    }
                }
                // 2.在主题对象中新增一个订阅者对象关联的连接 在订阅者对象中新增订阅主题
                topic->addSubscriber(subscriber);
                subscriber->addTopic(topic_name);
                return true;
            } 
            // 主题取消订阅
            void topicCancel(const BaseConnection::ptr &conn, const TopicRequest::ptr &top_req){
                // 1.根据订阅的主题找到主题对象 以及订阅者对象
                std::string topic_name=top_req->getTopicKey();
                Topic::ptr topic;
                Subscriber::ptr subscriber;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topics.find(topic_name);
                    if (topic_it != _topics.end()){
                        topic = topic_it->second;
                    }
                    auto sub_it = _subscribers.find(conn);
                    if (sub_it != _subscribers.end()){
                        subscriber = sub_it->second;
                    }
                }
                // 从主题对象中删除订阅者关联的连接 从订阅者对象中删除所订阅的主题名称
                if(topic){
                    topic->removeSubscriber(subscriber);
                }
                if(subscriber){
                    subscriber->removeTopic(topic_name);
                }
            }
            // 主题发布
            bool topicPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &top_req){
                Topic::ptr topic;
                std::string topic_name = top_req->getTopicKey();
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topics.find(topic_name);
                    if (topic_it == _topics.end()){
                        server_logger->ERROR("发布失败: 主题 '%s' 不存在", topic_name.c_str());
                        return false;
                    }
                    topic = topic_it->second;
                }
                topic->pushMessage(top_req);
                return true;
            }

            // 主题操作正确响应发送
            void topicResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &top_req, RCode rcode){
                auto top_rsp = MessageFactory::create<TopicResponse>();
                top_rsp->setMsgId(top_req->getMsgId());
                top_rsp->setMsgType(MsgType::RSP_TOPIC);
                top_rsp->setRCode(rcode);
                conn->send(top_rsp);
            }
            // 主题操作错误响应发送
            void errorResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &top_req, RCode rcode){
                auto top_rsp = MessageFactory::create<TopicResponse>();
                top_rsp->setMsgId(top_req->getMsgId());
                top_rsp->setMsgType(MsgType::RSP_TOPIC);
                top_rsp->setRCode(rcode);
                conn->send(top_rsp);
            }

        private:
            std::mutex _mutex;
            std::unordered_map<std::string, Topic::ptr> _topics;                     // 主题对象的集合
            std::unordered_map<BaseConnection::ptr, Subscriber::ptr> _subscribers;   // 订阅者对象的集合
        };
    }
}